/**
 * @Title: Producer.java
 * @Package com.study.kafka
 * @Description: TODO
 * Copyright: Copyright (c) 2011
 * Company:成都康赛电子科大信息技术有限责任公司
 * @author Comsys-zhang
 * @date 2019年5月21日 下午9:54:16
 * @version V1.0
 */

package com.study.spring.kafka;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;


/**
 * @author Comsys-zhang
 * @ClassName: Producer
 * @Description: kafka生产者
 * @date 2019年5月21日 下午9:54:16
 */
public class Producer {

    private static final Logger LOG = LoggerFactory.getLogger(Producer.class);

    private static final String BROKER_LIST = "192.168.1.4:9092";

    private static final String TOPIC = "zhangyi";

    private static KafkaProducer<String, String> producer = null;

    static {
        Properties config = initConfig();
        producer = new KafkaProducer<>(config);
    }

    /**
     * initConfig(这里用一句话描述这个方法的作用)
     *
     * @param @return 设定文件
     * @return Properties    返回类型
     * @throws
     * @Title: initConfig
     * @Description: 构建生产者配置
     */
    private static Properties initConfig() {
        Properties properties = new Properties();

        Properties props = new Properties();
        props.put("bootstrap.servers", BROKER_LIST);
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return properties;
    }

    public static void main(String[] args) {
        ProducerRecord<String, String> record = null;
        for (int i = 0; i < 1000; i++) {
            record = new ProducerRecord<String, String>(TOPIC, "value" + Math.random());

            producer.send(record, new Callback() {

                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (null != exception) {
                        LOG.error("send error {}", exception.getMessage());
                    } else {
                        LOG.info("offset:{},partition:{}", metadata.offset(), metadata.partition());
                    }

                }
            });
        }
        producer.close();
    }
}
